"""
Case Type   : HTAP行列融合特性
Case Name   : HTAP行列融合特性--不支持toast表，合理报错
Create At   : 2025/11/3
Owner       : sungang14
Description :
    1、建表
    2、插入数据
    3、查看toast表数据
    4、全表转换
    5、主备查询
    6、取消转换
    7、部分列转换
    8、主备查询
    9、取消转换
    10、指定分区转换
    11、主备查询
    12、取消转换
    13、指定分区部分列转换
    14、主备查询
    15、取消转换
    16、环境清理
Expect      :
    1、执行成功
    2、执行成功
    3、执行成功
    4、合理报错
    5、主备查询成功
    6、合理报错
    7、合理报错
    8、主备查询成功
    9、合理报错
    10、合理报错
    11、主备查询成功
    12、合理报错
    13、合理报错
    14、主备查询成功
    15、合理报错
    16、环境清理成功
History     :
"""

import os
import unittest
import time

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Logger import Logger
from testcase.utils.Constant import Constant
from yat.test import macro


class HTAPIntegerTypes(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.sta1_sh = CommonSH('Standby1DbUser')
        self.sta2_sh = CommonSH('Standby2DbUser')
        self.constant = Constant()
        status = self.pri_sh.get_db_cluster_status('status')
        self.log.info(status)
        if not self:
            self.pri_sh.restart_db_cluster()

        self.pg_hba_backup = None
        text = '----备份pg_hba.conf文件;expect:执行成功----'
        self.log.info(text)
        cmd = f'source {macro.DB_ENV_PATH};' \
              f'cp {macro.DB_INSTANCE_PATH}/pg_hba.conf {macro.DB_INSTANCE_PATH}/pg_hba.conf.backup.test'
        self.log.info(cmd)
        exec_res = self.pri_sh.node.sh(cmd).result()
        self.log.info(exec_res)

        text = '----配置主机信任连接;expect:执行成功----'
        self.log.info(text)
        cmd = f'source {macro.DB_ENV_PATH};' \
              f'gs_guc reload -N all -I all -h "host all all {self.pri_sh.node.db_host}/32 trust";' \
              f'gs_guc reload -N all -I all -h "host all all {self.sta1_sh.node.db_host}/32 trust";' \
              f'gs_guc reload -N all -I all -h "host all all {self.sta2_sh.node.db_host}/32 trust"'
        self.log.info(cmd)
        exec_res = self.pri_sh.node.sh(cmd).result()
        self.log.info(exec_res)
        self.table_name_1 = 't_htap_0021_1'
        self.table_name_2 = 't_htap_0021_2'
        self.table_name_3 = 't_htap_0021_3'

    def test_integer_types_htap(self):
        text = '----step1:建表;expect:执行成功----'
        self.log.info(text)
        sql_cmd = f'drop table if exists {self.table_name_1} cascade;drop table if exists {self.table_name_2} cascade;drop table if exists {self.table_name_3} cascade;'
        res = self.pri_sh.execut_db_sql(sql_cmd)
        self.assertEqual(res.count(self.constant.TABLE_DROP_SUCCESS), 3, f'执行失败:{text}')
        sql = f"""create table {self.table_name_1}(
                    c1 int,
                    c2 text not null
                    );"""
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertEqual(self.constant.TABLE_CREATE_SUCCESS, q_res, f'执行失败:{text}')
        sql = f"""create table {self.table_name_2}(
                        c1 int,
                        c2 text
                    )
                    partition by range (c1)
                    (
                        partition t_htap_0021_2_p1 values less than (50),
                        partition t_htap_0021_2_p2 values less than (1000)
                    );"""
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertEqual(self.constant.TABLE_CREATE_SUCCESS, q_res, f'执行失败:{text}')
        sql = f"""create table t_htap_0021_3(
                        c1 date,
                        c2 int,
                        c3 text
                    )partition by range (c1) subpartition by range(c2)
                    (
                        partition t_htap_0021_3_p1 values less than ('2024-01-10')
                        (
                            subpartition t_htap_0021_3_p1_a values less than(50),
                            subpartition t_htap_0021_3_p1_b values less than(100)
                        ),
                        partition t_htap_0021_3_p2 values less than ('2024-01-20')
                        (
                            subpartition t_htap_0021_3_p2_a values less than(50),
                            subpartition t_htap_0021_3_p2_b values less than(100)
                        ),
                        partition t_htap_0021_3_p3 values less than ('2024-02-01')
                        (
                            subpartition t_htap_0021_3_p3_a values less than(50),
                            subpartition t_htap_0021_3_p3_b values less than(100)
                        )
                    );"""
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertEqual(self.constant.TABLE_CREATE_SUCCESS, q_res, f'执行失败:{text}')

        text = '----step2:插入数据;expect:执行成功----'
        self.log.info(text)
        sql = f"insert into {self.table_name_1} values (generate_series(1,100), generate_series(1,100)||repeat('hello world', 100000));"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertEqual(f"INSERT 0 100", q_res, f'执行失败:{text}')
        sql = f"insert into {self.table_name_2} values (generate_series(1,100), generate_series(1,100)||repeat('hello world', 100000));"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertEqual(f"INSERT 0 100", q_res, f'执行失败:{text}')
        sql = f"insert into {self.table_name_3} select generate_series('2024-01-01'::timestamp, '2024-01-01'::timestamp + interval '30 days', '1 day'::interval), generate_series(1, 99), generate_series(1, 99)||repeat('熊猫竹竹爱吃竹子', 10000);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertEqual(f"INSERT 0 3069", q_res, f'执行失败:{text}')

        text = '----step3:查看toast表;expect:执行成功----'
        self.log.info(text)
        sql = f"select b.relname from pg_class a, pg_class b where a.reltoastrelid=b.relfilenode and a.relname='{self.table_name_1}';"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(f"pg_toast_", q_res.splitlines()[-2].strip(), f'执行失败:{text}')
        sql = f"select a.relname, b.relname from pg_partition a, pg_class b where a.reltoastrelid=b.oid order by a.relname;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(f"pg_toast_part_", q_res, f'执行失败:{text}')
        self.assertIn(f"t_htap_0021_", q_res, f'执行失败:{text}')

        text = '----step4:全表转换;expect:执行合理报错----'
        self.log.info(text)
        sql = f"alter table {self.table_name_1} imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assert1_msg = f"col [c2] is unknow, not supported by imcstore"
        self.assert3_msg = f"col [c3] is unknow, not supported by imcstore"
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_2} imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_3} imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert3_msg, q_res, f'执行失败:{text}')

        text = '----step5:主备查询;expect:查询成功----'
        self.log.info(text)
        sql = f"select count(*) from {self.table_name_1} where c1 between 20 and 50;" \
              f"select count(*) from {self.table_name_2} where c1 between 20 and 50;" \
              f"select count(*) from {self.table_name_3} where c2 between 20 and 50;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(f'主节点查询结果:\n{q_res}')
        self.assertEqual(961, int(q_res.splitlines()[-2].strip()), f'执行失败:{text}')
        self.assertEqual(31, int(q_res.splitlines()[-7].strip()), f'执行失败:{text}')
        self.assertEqual(31, int(q_res.splitlines()[-12].strip()), f'执行失败:{text}')
        sql = f"set enable_imcsscan to on;" \
              f"select count(*) from {self.table_name_1} where c1 between 20 and 50;" \
              f"select count(*) from {self.table_name_2} where c1 between 20 and 50;" \
              f"select count(*) from {self.table_name_3} where c2 between 20 and 50;"
        q_res = self.sta1_sh.execut_db_sql(sql)
        self.log.info(f'备节点查询结果:\n{q_res}')
        self.assertEqual(961, int(q_res.splitlines()[-2].strip()), f'执行失败:{text}')
        self.assertEqual(31, int(q_res.splitlines()[-7].strip()), f'执行失败:{text}')
        self.assertEqual(31, int(q_res.splitlines()[-12].strip()), f'执行失败:{text}')

        text = '----step6:取消转换;expect:合理报错----'
        self.log.info(text)
        sql = f"alter table {self.table_name_1} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assert4_msg = f"rel not populated, no need to be unpopulate"
        self.assertIn(self.assert4_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_2} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert4_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_3} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert4_msg, q_res, f'执行失败:{text}')

        text = '----step7:部分列转换;expect:执行合理报错----'
        self.log.info(text)
        sql = f"alter table {self.table_name_1} imcstored(c2);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_2} imcstored(c2);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_3} imcstored(c2, c3);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert3_msg, q_res, f'执行失败:{text}')

        text = '----step8:主备查询;expect:查询成功----'
        self.log.info(text)
        sql = f"select count(*) from {self.table_name_1} where c1 between 20 and 50;" \
              f"select count(*) from {self.table_name_2} where c1 between 20 and 50;" \
              f"select count(*) from {self.table_name_3} where c2 between 20 and 50;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(f'主节点查询结果:\n{q_res}')
        self.assertEqual(961, int(q_res.splitlines()[-2].strip()), f'执行失败:{text}')
        self.assertEqual(31, int(q_res.splitlines()[-7].strip()), f'执行失败:{text}')
        self.assertEqual(31, int(q_res.splitlines()[-12].strip()), f'执行失败:{text}')
        sql = f"set enable_imcsscan to on;" \
              f"select count(*) from {self.table_name_1} where c1 between 20 and 50;" \
              f"select count(*) from {self.table_name_2} where c1 between 20 and 50;" \
              f"select count(*) from {self.table_name_3} where c2 between 20 and 50;"
        q_res = self.sta1_sh.execut_db_sql(sql)
        self.log.info(f'备节点查询结果:\n{q_res}')
        self.assertEqual(961, int(q_res.splitlines()[-2].strip()), f'执行失败:{text}')
        self.assertEqual(31, int(q_res.splitlines()[-7].strip()), f'执行失败:{text}')
        self.assertEqual(31, int(q_res.splitlines()[-12].strip()), f'执行失败:{text}')

        text = '----step9:取消转换;expect:合理报错----'
        self.log.info(text)
        sql = f"alter table {self.table_name_1} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert4_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_2} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert4_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_3} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert4_msg, q_res, f'执行失败:{text}')

        text = '----step10:指定分区转换;expect:执行合理报错----'
        self.log.info(text)
        sql = f"alter table {self.table_name_2} modify partition t_htap_0021_2_p1 imcstored(c2);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_3} modify partition t_htap_0021_3_p1 imcstored(c2, c3);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert3_msg, q_res, f'执行失败:{text}')

        text = '----step11:主备查询;expect:查询成功----'
        self.log.info(text)
        sql = f"select count(*) from t_htap_0021_2 partition (t_htap_0021_2_p1);" \
              f"select count(*) from t_htap_0021_3 partition (t_htap_0021_3_p1);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(f'主节点查询结果:\n{q_res}')
        self.assertEqual(891, int(q_res.splitlines()[-2].strip()), f'执行失败:{text}')
        self.assertEqual(49, int(q_res.splitlines()[-7].strip()), f'执行失败:{text}')
        sql = f"set enable_imcsscan to on;" \
              f"select count(*) from t_htap_0021_2 partition (t_htap_0021_2_p1);" \
              f"select count(*) from t_htap_0021_3 partition (t_htap_0021_3_p1);"
        q_res = self.sta1_sh.execut_db_sql(sql)
        self.log.info(f'备节点查询结果:\n{q_res}')
        self.assertEqual(891, int(q_res.splitlines()[-2].strip()), f'执行失败:{text}')
        self.assertEqual(49, int(q_res.splitlines()[-7].strip()), f'执行失败:{text}')

        text = '----step12:取消转换;expect:合理报错----'
        self.log.info(text)
        sql = f"alter table {self.table_name_2} modify partition t_htap_0021_2_p1 unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(f"not populated", q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_3} modify partition t_htap_0021_3_p1 unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(f"not populated", q_res, f'执行失败:{text}')

        text = '----step13:指定分区部分列转换;expect:执行合理报错----'
        self.log.info(text)
        sql = f"alter table {self.table_name_2} modify partition t_htap_0021_2_p1 imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_3} modify partition t_htap_0021_3_p1 imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert3_msg, q_res, f'执行失败:{text}')

        text = '----step14:主备查询;expect:查询成功----'
        self.log.info(text)
        sql = f"select count(*) from t_htap_0021_2 partition (t_htap_0021_2_p1);" \
              f"select count(*) from t_htap_0021_3 partition (t_htap_0021_3_p1);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(f'主节点查询结果:\n{q_res}')
        self.assertEqual(891, int(q_res.splitlines()[-2].strip()), f'执行失败:{text}')
        self.assertEqual(49, int(q_res.splitlines()[-7].strip()), f'执行失败:{text}')
        sql = f"set enable_imcsscan to on;" \
              f"select count(*) from t_htap_0021_2 partition (t_htap_0021_2_p1);" \
              f"select count(*) from t_htap_0021_3 partition (t_htap_0021_3_p1);"
        q_res = self.sta1_sh.execut_db_sql(sql)
        self.log.info(f'备节点查询结果:\n{q_res}')
        self.assertEqual(891, int(q_res.splitlines()[-2].strip()), f'执行失败:{text}')
        self.assertEqual(49, int(q_res.splitlines()[-7].strip()), f'执行失败:{text}')

        text = '----step15:取消转换;expect:合理报错----'
        self.log.info(text)
        sql = f"alter table {self.table_name_2} modify partition t_htap_0021_2_p1 unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(f"not populated", q_res, f'执行失败:{text}')
        sql = f"alter table {self.table_name_3} modify partition t_htap_0021_3_p1 unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(f"not populated", q_res, f'执行失败:{text}')

    def tearDown(self):
        text = '----step16:清理环境;expect:执行成功----'
        self.log.info(text)
        sql_cmd = f'drop table if exists {self.table_name_1} cascade;drop table if exists {self.table_name_2} cascade;drop table if exists {self.table_name_3} cascade;'
        res = self.pri_sh.execut_db_sql(sql_cmd)
        self.assertEqual(res.count(self.constant.TABLE_DROP_SUCCESS), 3, f'执行失败:{text}')

        text = '----恢复pg_hba.conf文件;expect:执行成功----'
        self.log.info(text)
        cmd = f'source {macro.DB_ENV_PATH};' \
              f'cp {macro.DB_INSTANCE_PATH}/pg_hba.conf.backup.test {macro.DB_INSTANCE_PATH}/pg_hba.conf;' \
              f'gs_om -t restart'
        self.log.info(cmd)
        exec_res = self.pri_sh.node.sh(cmd).result()
        self.log.info(exec_res)
        time.sleep(30)
        cmd = f'source {macro.DB_ENV_PATH};gs_om -t status --detail'
        status_res = self.pri_sh.node.sh(cmd).result()
        self.log.info(f'数据库集群状态:\n{status_res}')
        cmd = f'source {macro.DB_ENV_PATH};rm -f {macro.DB_INSTANCE_PATH}/pg_hba.conf.backup.test'
        self.pri_sh.node.sh(cmd).result()
        self.log.info('pg_hba.conf备份文件已清理')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
